package com.ry.flink.watermark;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

public class IndicatorProcessMainByTransportTypeWaterMark implements AssignerWithPeriodicWatermarks<Tuple3<Long,String,String>> {

    long currentMaxTimestamp = 0L;
    final long maxOrderness = 5000000000L;

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOrderness);
    }

    /**
     * 指定eventTime
     * */
    @Override
    public long extractTimestamp(Tuple3<Long, String, String> element, long previousElementTimestamp) {
        long current_timestamp = element.f0;
        currentMaxTimestamp = Math.max(current_timestamp,currentMaxTimestamp);
        return current_timestamp;
    }
}
